package com.jscloud.spark.pvuv

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext, rdd}

object VisitorTopN {
  def main(args: Array[String]): Unit = {
    //程序入口类 SparkContext
    val sparkConf: SparkConf = new SparkConf().setAppName("VisitorTopN").setMaster("local[*]")
    val sc: SparkContext = new SparkContext(sparkConf)
    sc.setLogLevel("WARN")
    val dataRdd: RDD[String] = sc.textFile("file:///D:\\JSProjects\\jsCloud-bigdata-app\\sparkapp\\src\\main\\resources\\access.log")

    //过滤一些 404 的请求
    val filterRdd: RDD[String] = dataRdd.filter((x => x.split(" ").length > 10))

    //获取所有的URL
    val urlRdd = filterRdd.map(x => x.split(" ")(10))

    val fullUrlRdd = urlRdd.filter(x => x.contains("http"))

    val urlAndOne: RDD[(String, Int)] = fullUrlRdd.map(x => (x, 1))

    val result = urlAndOne.reduceByKey(_ + _)

    val sortRdd = result.sortBy(x => x._2, false)

    val top5 = sortRdd.take(5)

    top5.foreach(println(_))

    sc.stop()

  }

}
